 
/* 
 * Copyright (C) Ignacio Lucas
 */
 
/*
 *  worker, hecho por Ignacio Lucas
 *
 *  Worker que implementa el modelo productor-consumidor, donde cada worker puede poner o tomar tareas de una cola, 
 *  cuyo acceso esta serializado mediante mecanismos de semaforos para evitar race-conditions.
 *
 *  Ignacio Lucas <ignacio.lucas@gmail.com>
 *
 */
#include "common.h"
#include "worker.h"
#include "priority_queue.h"
#include <pthread.h>
#include <limits.h>

#include "common_objects_pool.h"

// Mutex & condition usados para acceso sincronizado entre workers a la cola de tareas.
static pthread_mutex_t jobs_mutex = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t cond_ready_jobs = PTHREAD_COND_INITIALIZER;

// Mutex para acceso sincronizado a funciones por parte de los workers.
static pthread_mutex_t exclusive_call_mutex = PTHREAD_MUTEX_INITIALIZER;

// mutex para acceso sincronizado en los distintos pools del common_objects_pool
static pthread_mutex_t pool_mutex[MAX_POOLS]={
PTHREAD_MUTEX_INITIALIZER,PTHREAD_MUTEX_INITIALIZER,PTHREAD_MUTEX_INITIALIZER,PTHREAD_MUTEX_INITIALIZER,
PTHREAD_MUTEX_INITIALIZER,PTHREAD_MUTEX_INITIALIZER,PTHREAD_MUTEX_INITIALIZER,PTHREAD_MUTEX_INITIALIZER,
PTHREAD_MUTEX_INITIALIZER,PTHREAD_MUTEX_INITIALIZER,PTHREAD_MUTEX_INITIALIZER,PTHREAD_MUTEX_INITIALIZER,
PTHREAD_MUTEX_INITIALIZER,PTHREAD_MUTEX_INITIALIZER,PTHREAD_MUTEX_INITIALIZER,PTHREAD_MUTEX_INITIALIZER,
PTHREAD_MUTEX_INITIALIZER,PTHREAD_MUTEX_INITIALIZER,PTHREAD_MUTEX_INITIALIZER,PTHREAD_MUTEX_INITIALIZER,
PTHREAD_MUTEX_INITIALIZER,PTHREAD_MUTEX_INITIALIZER,PTHREAD_MUTEX_INITIALIZER,PTHREAD_MUTEX_INITIALIZER};

// clave para acceder a datos que necesite la aplicacion resguardar.
static pthread_key_t key_user_data;

// estructura que registra la funcion que ejecuta el worker junto con sus argumentos (parametros).
typedef struct s_worker_startup_data
{
	int (*main_worker_job)(void *);
	void *args;

}worker_startup_data;


// estructura asociada a un worker.
typedef struct s_worker
{
		// hilo de ejecucion (POSIX Thread) asociado al worker.
		pthread_t executor;
	
		// datos de inicializacion para el worker.
		worker_startup_data startup_data;
		
		// cola donde los workers dejan y toman tareas.
		priority_queue_t queue;
		
		// identificador del worker
		int id;
		
		// status. 1: worker iniciado, 0: worker detenido.
		int status;

		// tamanio del stack de ejecucion.
		size_t stack_size;
		
} worker;



// inicializacion del modulo. Debe ser llamado una sola vez al comienzo del programa.
int worker_init_module()
{
	// solamente inicializamos la key para los datos de usuario.
	return pthread_key_create(&key_user_data,NULL);
}

void worker_destroy_module()
{
	pthread_key_delete(key_user_data);
}


void *main_worker_job_test(void *args)
{
	while(1>0);
	return 0;
}

// set y get de los datos de usuario.
inline int worker_set_user_data(void *user_data)
{
	return pthread_setspecific(key_user_data,user_data);
}

inline void *worker_get_user_data(void)
{
	return pthread_getspecific(key_user_data);
}


// inicializacion del worker. Aloca memoria para el objeto worker.
worker_t worker_init(int (*main_worker_job)(void *), void *args, priority_queue_t queue, size_t stack_size, int id)
{
		worker *wk = NULL;

		wk = (worker *)malloc(sizeof(worker));
		if (wk == NULL) return NULL;
		
		wk->queue=queue;
		wk->id = id;
		wk->status=0;
		wk->stack_size = stack_size;
		
		wk->startup_data.main_worker_job = main_worker_job;
		wk->startup_data.args = args;

	  return (worker_t)wk;
}

// wrapper para registrar el worker_id antes de lanzar la aplicacion.
static int worker_start_wrapper(worker *wk)
{
	char worker_id[20];
	sprintf(worker_id,"worker_%d",wk->id);
	logger_register_worker(worker_id);

	int ret = wk->startup_data.main_worker_job(wk->startup_data.args);
	return ret;
}

// comienza la ejecucion del hilo, ejecutando la funcion que se indico en worker_init
int worker_start(worker_t wkt)
{
	int ret = 0;
	pthread_attr_t attr;
	worker *wk = NULL;
	size_t stack_size=(size_t)0, default_stack_size=(size_t)0;
	
	wk=(worker *)wkt;

	// indico que este hilo es JOINABLE (significa que otro hilo va a esperar la detencion de este hilo).
	pthread_attr_init(&attr);
	pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

	// Obtengo el tamanio default de stack.
	ret=pthread_attr_getstacksize(&attr, &default_stack_size);

	// Asignamos el tamanio preferido de stack.
	stack_size=wk->stack_size;
	ret=pthread_attr_setstacksize(&attr, stack_size);
	if (ret != 0)
	{
		logger_core_error( "No se pudo setear el tamanio de stack para pthread en '%d', usando valor default %d", stack_size, default_stack_size);
  	}
	//ret = pthread_create(&(wk->executor), &attr, (void *)wk->startup_data.main_worker_job, wk->startup_data.args);
	ret = pthread_create(&(wk->executor), &attr,(void *(*) (void *)) worker_start_wrapper, wk);
	
	// asigno el status del hilo de ejecucion en funcion del resultado de pthread_create.
	wk->status=((ret==0)?1:0);
	return ret;
}

int worker_get_id(worker_t wkt)
{
	worker *wk = (worker *)wkt;
	return wk->id;
}

int worker_push_job_priority(worker_t w_t, void *job, int priority)
{
	worker *wk=(worker *)w_t;
	pthread_mutex_lock(&jobs_mutex);
	priority_queue_push(wk->queue,job,priority);
	(void)pthread_cond_signal(&cond_ready_jobs);
	pthread_mutex_unlock(&jobs_mutex);
	return 0;
}

int worker_push_job(worker_t w_t, void *job)
{
	return worker_push_job_priority(w_t,job,MASHUP_SERVER_PRIORITY_QUEUE_NORMAL);
}

int worker_push_job_vector_priority(worker_t w_t, void **jobs, int size,int priority)
{
	worker *wk=(worker *)w_t;
	pthread_mutex_lock(&jobs_mutex);
	priority_queue_push_vector(wk->queue,jobs, size,priority);
	(void)pthread_cond_signal(&cond_ready_jobs);
	pthread_mutex_unlock(&jobs_mutex);
	return 0;

}

int worker_push_job_vector(worker_t w_t, void **jobs, int size)
{
	return worker_push_job_vector_priority(w_t,jobs,size,MASHUP_SERVER_PRIORITY_QUEUE_NORMAL);
}

void *worker_pull_job(worker_t w_t)
{
	worker *wk=(worker *)w_t;
	void *job = NULL;
	
	errno=0;
	pthread_mutex_lock(&jobs_mutex);
	while ((job = priority_queue_pull(wk->queue)) == NULL)
	{
		(void)pthread_cond_wait(&cond_ready_jobs, &jobs_mutex); 
	}
	pthread_mutex_unlock(&jobs_mutex);
	return job;
}

void *worker_pick_job(worker_t w_t)
{
	worker *wk=(worker *)w_t;
	void *job = NULL;
	pthread_mutex_lock(&jobs_mutex);
	job=priority_queue_pull(wk->queue);
	pthread_mutex_unlock(&jobs_mutex);
	return job;
}

int worker_destroy(worker_t w)
{
	int id=-1,ret=0;
	worker *wk = (worker *)w;
	if (wk != NULL)
	{
		id=worker_get_id(w);
		if (wk->status == 1)
		{
			pthread_cancel((wk->executor));
		}
		else
		{
			ret=1;				
		}
		wk->queue=NULL;
		free(wk);
		wk=NULL;
	}
	else
	{
		ret=2;
	}
	logger_core_info( "%s [%d].ret=%d", "finalizacion de worker", id, ret);
	return ret;
}

inline int worker_object_pool_put_exclusive(int pool_type, void *p, object_pool *pool)
{
	int ret=0;
	pthread_mutex_lock(&(pool_mutex[pool_type]));
	ret=object_pool_put_or_free(pool, p);
	pthread_mutex_unlock(&(pool_mutex[pool_type]));
	return ret;
}

inline void* worker_object_pool_get_exclusive(int pool_type, object_pool *pool)
{
	void *ret=NULL;
	pthread_mutex_lock(&(pool_mutex[pool_type]));
	ret=object_pool_pick_or_alloc(pool);
	pthread_mutex_unlock(&(pool_mutex[pool_type]));
	return ret;
}

inline int worker_call_exclusive(fn_call_exclusive func, void *arg1, void *arg2)
{
	int ret=0;
	pthread_mutex_lock(&(exclusive_call_mutex));
	ret=func(arg1, arg2);
	pthread_mutex_unlock(&(exclusive_call_mutex));
	return ret;
}

